package removeDanglingPage;

import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import mapred.AbstractReducer;
import mapred.util.Bytes;

public class RemoveDP_GetReady_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	byte[] page = {};
	byte[] outdegree = {};
	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		
		ReadBytes k = new ReadBytes(key);
		byte[] p = new byte[16];
		p = k.nextUrl();
		k.next();
		if(!Arrays.equals(page, p)) {
			reporter.incrCounter("Remove", "sum", 1);
			if(key.getLength() < 32) {
				reporter.incrCounter("Remove", "num", 1);
				page = Arrays.copyOf(p, p.length);
				outdegree = k.nextBytes(4);
			} else {
				byte[] pp = new byte[16];
				pp = k.nextUrl();
				while(values.hasNext()) {
					reporter.incrCounter("Remove", "double", 1);
					output.collect(new BytesWritable(Bytes.concat(p, Bytes.Integer2Bytes(0))), 
							new BytesWritable(pp));
					byte[] url0 = Bytes.concat(pp, new byte[] {0});
					output.collect(new BytesWritable(Bytes.concat(url0, Bytes.Integer2Bytes(-1))), 
							new BytesWritable(new byte[0]));
					values.next();
				}
			}
		} else {
			reporter.incrCounter("Remove", "fff", 1);
			byte[] dp = k.nextUrl();
			while(values.hasNext()) {
				reporter.incrCounter("Remove", "has", 1);
				byte[] url1 = Bytes.concat(page, new byte[] {1});
				output.collect(new BytesWritable(Bytes.concat(url1, outdegree)), 
						new BytesWritable(dp));
				values.next();
			}
			
		}
	}

//	private String page = "";
//	
//	private int outdegree = 0;
//	
//	@Override
//	public void reduce(Text key, Iterator<Text> values,
//			OutputCollector<Text, Text> output, Reporter reporter)
//			throws IOException {
//		// TODO Auto-generated method stub
//		String keyStr = key.toString();
//		String[] t = keyStr.split(" ");
//		if (page.equals(t[0])) {
//			while(values.hasNext()) {
//				output.collect(new Text(page + " " +outdegree), new Text(t[1]));
//				values.next();
//			}
//			
//		} else {
//			
//			try {
//				if(Integer.parseInt(t[1]) > 0) {
//					page = t[0];
//					outdegree = Integer.parseInt(t[1]);
//				} 
//			} catch(Exception e) {
//				while(values.hasNext()) {
//					output.collect(new Text(t[0] + " " + 0), new Text(t[1]));
//					output.collect(new Text(t[1] + " " + (-1)), new Text(""));
//					values.next();
//				}
//			}
//		}
//	}

}
